package priv.bestbeat.reportcenter.core.mq.rocket;

import lombok.NoArgsConstructor;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import priv.bestbeat.reportcenter.core.monitor.Observable;
import priv.bestbeat.reportcenter.core.mq.MessageProxy;

import javax.annotation.Resource;
import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;

/**
 * @author 张渠钦
 * @date 2022/4/24 17:16
 */
@NoArgsConstructor
public class RocketMQProxy extends MessageProxy<RocketMessage> {

    @Resource
    private MQProducer mqProducer;

    public RocketMQProxy(MQProducer mqProducer) {
        this.mqProducer = mqProducer;
    }

    @Override
    public RocketMessage send(RocketMessage message) throws Exception {
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
             ObjectOutputStream oos = new ObjectOutputStream(baos);) {
            oos.writeObject(message.getPayload());
            Message msg = new Message(message.getTopic(),baos.toByteArray());
            mqProducer.send(msg);
        }

        message.sendDone();

        return message;
    }

}
